package cn.oowoo.rabbitmq.rpc;

import cn.oowoo.rabbitmq.config.RabbitConfig;
import cn.oowoo.rabbitmq.config.RabbitExchangeTypeEnum;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeoutException;

/**
 * @Author ZhangBin
 * @Date 2021/5/8 13:48
 * @Description: 简单消息发送-生产者
 *  测试额外参数传递、消息过期时间等
 **/
public class RpcProducer {

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        Connection connection = RabbitConfig.createConnection();
        // 创建消息通道
        Channel channel = connection.createChannel();

        //模拟一批数据进行PRC调用
        for(int i=10;i<15;i++){
            for(int j=1;j<5;j++){
                String msg = "{\"a\":"+i+",\"b\":"+j+"}";//计算a+b的值
                //设置消息属性的replyTo和correlationId
                AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
                        .correlationId(java.util.UUID.randomUUID().toString())
                        .replyTo(RabbitConfig.QUEUE_RPC_RESPONSE_NAME)//回调队列
                        .contentType("application/json")
                        .build();
                channel.basicPublish(RabbitConfig.EXCHANGE_RPC_NAME, RabbitConfig.ROUTING_RPC_KEY, properties, msg.getBytes());
                System.out.println("发送消息msg---> "+ msg);
            }
        }


        //从匿名内部类中获取返回值
        final BlockingQueue<String> response = new ArrayBlockingQueue<String>(50);
        // 创建消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                response.offer(new String(body, "UTF-8"));
            }
        };
        // 开始获取消息
        channel.basicConsume(RabbitConfig.QUEUE_RPC_RESPONSE_NAME, true, consumer);//监听回调队列
        while (true){
            String take = response.take();
            System.out.println("获取RPC远程调用消息："+take);
            Thread.sleep(100);
        }





        //消息发送完毕关闭信道，关闭链接(不关闭连接，要监听回调队列中的消息)
        //channel.close();
        //connection.close();
    }
}
